#ifndef __M_BROKER_H__
    #define __M_BROKER_H__
#include "protobuf/codec/codec.h"
#include "protobuf/codec/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "../mqcommon/mq_loger.hpp"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "mq_connection.hpp"
#include "mq_host.hpp"
#include "mq_consumer.hpp"
#include <iostream>
#include <functional>

namespace mq{

class Server{
public:
typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
#define DBFILE "/meta.db"
#define HOSTNAME "MyVirtualHost"

Server(uint16_t port,const std::string& basedir):_server(&_baseloop,muduo::net::InetAddress("0.0.0.0",port),"TcpServer"),
    _dispatcher(std::bind(&Server::onUnknownMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)),
    _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage,&_dispatcher,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3))),
    _virtual_host(std::make_shared<mq::VirtualHost>(HOSTNAME,basedir,basedir + DBFILE)),
    _consuer_manager(std::make_shared<mq::ConsumerManager>()),
    _connection_manager(std::make_shared<mq::ConnectionManager>()),
    _threadpool_manager(std::make_shared<mq::threadpool>())
    {
      //其中对于消费者来说，他内部并不会自动的回复历史消费者队列，所以需要获取虚拟机中的队列消息，并初始化给消费者队列
      mq::QueueMap qm = _virtual_host->allQueues();
      //遍历队列map，获取队列名称，创建消费者队列，完成队列回复
      for(auto& q : qm){
          _consuer_manager->initQueueConsumer(q.first);
      }
//给收到的请求，注册回调函数
      _dispatcher.registerMessageCallback<openChannelRequest>(std::bind(&Server::onOpenChannel,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<closeChannelRequest>(std::bind(&Server::onCloseChannel,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      
      _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&Server::onDeclareExchange,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&Server::onDeleteExchange,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      
      _dispatcher.registerMessageCallback<declareQueueRequest>(std::bind(&Server::onDeclareQueue,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<deleteQueueRequest>(std::bind(&Server::onDeleteQueue,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      
      _dispatcher.registerMessageCallback<queueBindRequest>(std::bind(&Server::onQueueBind,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<queueUnBindRequest>(std::bind(&Server::onQueueUnBind,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      
      
      
      _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&Server::onBasicPublish,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&Server::onBasicAck,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<basicConsumeRequest>(std::bind(&Server::onBasicConsume,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
      _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&Server::onBasicCancel,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));


      _server.setConnectionCallback(std::bind(&Server::onConnection,this,std::placeholders::_1));
      _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage,_codec.get(),std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
    }

    void start()
    {
      _server.start();
      _baseloop.loop();//开启循环监听
    }
private:
//未知事件就绪：
  void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                        const MessagePtr& message,
                        muduo::Timestamp)
  {
    LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
    conn->shutdown();
  }

//打开信道
  void onOpenChannel(const muduo::net::TcpConnectionPtr& conn,const openChannelRequestPtr& req,muduo::Timestamp){
    //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("打开信道时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 调用连接中的相关函数
    return mconn->openChannel(req);
  }

//关闭信道
  void onCloseChannel(const muduo::net::TcpConnectionPtr& conn,const closeChannelRequestPtr& req,muduo::Timestamp){
    //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("关闭信道时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //调用连接中的相关函数
    return mconn->closeChannel(req);
  }





//声明交换机
  void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn,const declareExchangeRequestPtr& req,muduo::Timestamp){
    //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("声明交换机时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 获取信道
    Channel::ptr cp = mconn->getChannel(req->cid());
    if(cp.get() == nullptr){
       DLOG("声明交换机时 没有找到信道!");
       return;
    }
    //3. 根据信道调用相关接口
    return cp->declareExchange(req);
  }

//删除交换机
  void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn,const deleteExchangeRequestPtr& req,muduo::Timestamp){
    //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("删除交换机时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 获取信道
    Channel::ptr cp = mconn->getChannel(req->cid());
    if(cp.get() == nullptr){
       DLOG("删除交换机时 没有找到信道!");
       return;
    }
    //3. 根据信道调用相关接口
    return cp->deleteExchange(req);
  }






//声明队列
  void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn,const declareQueueRequestPtr& req,muduo::Timestamp){
    //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("声明队列时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 获取信道
    Channel::ptr cp = mconn->getChannel(req->cid());
    if(cp.get() == nullptr){
       DLOG("声明队列时 没有找到信道!");
       return;
    }
    //3. 根据信道调用相关接口
    return cp->declareQueue(req);
  }

//删除队列
  void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn,const deleteQueueRequestPtr& req,muduo::Timestamp){
   //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("删除队列时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 获取信道
    Channel::ptr cp = mconn->getChannel(req->cid());
    if(cp.get() == nullptr){
       DLOG("删除队列时 没有找到信道!");
       return;
    }
    //3. 根据信道调用相关接口
    return cp->deleteQueue(req);
  }







//队列绑定
  void onQueueBind(const muduo::net::TcpConnectionPtr& conn,const queueBindRequestPtr& req,muduo::Timestamp){
   //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("队列绑定时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 获取信道
    Channel::ptr cp = mconn->getChannel(req->cid());
    if(cp.get() == nullptr){
       DLOG("队列绑定时 没有找到信道!");
       return;
    }
    //3. 根据信道调用相关接口
    return cp->queueBind(req);
  }

//队列解除绑定
  void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn,const queueUnBindRequestPtr& req,muduo::Timestamp){
       //1. 根据conn获取相关连接
    Connection::ptr mconn =  _connection_manager->getConnection(conn);
    if(mconn.get() == nullptr){
      DLOG("队列解除绑定时，没有找到连接对应的Connection对象！");
      conn->shutdown();//关闭连接
      return;
    }
    //2. 获取信道
    Channel::ptr cp = mconn->getChannel(req->cid());
    if(cp.get() == nullptr){
       DLOG("队列解除绑定时 没有找到信道!");
       return;
    }
    //3. 根据信道调用相关接口
    return cp->queueUnBind(req);
  }





//消息发布
  void onBasicPublish(const muduo::net::TcpConnectionPtr& conn,const basicPublishRequestPtr& req,muduo::Timestamp){
       //1. 根据conn获取相关连接
      Connection::ptr mconn =  _connection_manager->getConnection(conn);
      if(mconn.get() == nullptr){
        DLOG("消息发布时，没有找到连接对应的Connection对象！");
        conn->shutdown();//关闭连接
        return;
      }
      //2. 获取信道
      Channel::ptr cp = mconn->getChannel(req->cid());
      if(cp.get() == nullptr){
        DLOG("消息发布时 没有找到信道!");
        return;
      }
      //3. 根据信道调用相关接口
      return cp->basicPublish(req);
  }


//消息确认
  void onBasicAck(const muduo::net::TcpConnectionPtr& conn,const basicAckRequestPtr& req,muduo::Timestamp){
       //1. 根据conn获取相关连接
      Connection::ptr mconn =  _connection_manager->getConnection(conn);
      if(mconn.get() == nullptr){
        DLOG("消息确认时，没有找到连接对应的Connection对象！");
        conn->shutdown();//关闭连接
        return;
      }
      //2. 获取信道
      Channel::ptr cp = mconn->getChannel(req->cid());
      if(cp.get() == nullptr){
        DLOG("消息确认时 没有找到信道!");
        return;
      }
      //3. 根据信道调用相关接口
      return cp->basicAck(req);
  }

//订阅消息
  void onBasicConsume(const muduo::net::TcpConnectionPtr& conn,const basicConsumeRequestPtr& req,muduo::Timestamp){
      DLOG("接收到订阅请求！,订阅者标识：%s",req->consumer_tag().c_str());
       //1. 根据conn获取相关连接
      Connection::ptr mconn =  _connection_manager->getConnection(conn);
      if(mconn.get() == nullptr){
        DLOG("订阅消息时，没有找到连接对应的Connection对象！");
        conn->shutdown();//关闭连接
        return;
      }
      //2. 获取信道
      Channel::ptr cp = mconn->getChannel(req->cid());
      if(cp.get() == nullptr){
        DLOG("订阅消息时 没有找到信道!");
        return;
      }
      // DLOG("开始调用相关信道中的订阅接口！");
      //3. 根据信道调用相关接口
      return cp->basicConsume(req);
  }

//取消消息
  void onBasicCancel(const muduo::net::TcpConnectionPtr& conn,const basicCancelRequestPtr& req,muduo::Timestamp){
       //1. 根据conn获取相关连接
      Connection::ptr mconn =  _connection_manager->getConnection(conn);
      if(mconn.get() == nullptr){
        DLOG("取消消息时，没有找到连接对应的Connection对象！");
        conn->shutdown();//关闭连接
        return;
      }
      //2. 获取信道
      Channel::ptr cp = mconn->getChannel(req->cid());
      if(cp.get() == nullptr){
        DLOG("取消消息时 没有找到信道!");
        return;
      }
      //3. 根据信道调用相关接口
      return cp->basicCancel(req);
  }




  void onConnection(const muduo::net::TcpConnectionPtr &conn){
        if(!conn->connected())
        {
            std::cout << "连接关闭" << std::endl;
            //关闭连接，也要删除相关连接
          _connection_manager->delConnection(conn);

        }
        else
        {
          std::cout << "新连接成功" << std::endl;
          //连接建立成功后创建一个相关的连接
          _connection_manager->newConnection(_virtual_host,_consuer_manager,_codec,conn,_threadpool_manager);
        }   
  }

private:
    ProtobufDispatcher _dispatcher;
    muduo::net::EventLoop _baseloop;
    muduo::net::TcpServer _server;
    mq::ProtobufCodecPtr _codec;

    mq::VirtualHost::ptr _virtual_host;//虚拟机管理句柄
    mq::ConsumerManager::ptr _consuer_manager;//消费者管理句柄
    mq::ConnectionManager::ptr _connection_manager;//连接管理对象
    mq::threadpool::ptr _threadpool_manager;//任务池（线程池）

};

}

#endif